// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#include <cxxopts.hpp>

#include "coin/proto/coin_market_enums.pb.h"
#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/base/feed_convert_papyrus.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/symbology/product_encyclopedia.h"
#include "coin2/exchange/base/symbology/product_info.h"
#include "coin2/strategy/strategy.h"
#include "corridor/format_util/ProtoHelper.h"
#include "papyrus/presto_feed_dataset/DatasetRecorderManager.h"

using namespace coin2::base::config;
using namespace coin2::exchange::feed;
using namespace coin2::app;

using ::coin2::base::ConvertFeedMessage;
using ::coin2::exchange::base::feed::FeedUpdate;
using ::coin2::exchange::base::feed_util::Index;
using ::coin2::exchange::base::symbology::IProductInfo;

using papyrus::presto_feed_dataset::DatasetRecorderManager;
using papyrus::presto_feed_dataset::DatasetRecorderManagerConfigProto;

std::unique_ptr<DatasetRecorderManager> CreateDatasetRecorderManager(
    const std::string& exchange,
    const std::string& normalized_symbol,
    const std::string& native_symbol,
    const std::string& date,
    const DatasetRecorderManagerConfigProto& config,
    const std::string& filename_suffix) {
  papyrus::presto_feed_dataset::HeaderProto header_proto;
  header_proto.set_symbol(normalized_symbol);
  header_proto.set_tick_size(1);
  header_proto.set_money_multiplier(1);
  papyrus::presto_feed_dataset::FeedRowHeader header(header_proto);
  header.key().exchange = exchange;
  header.key().symbol = normalized_symbol;
  header.key().native_symbol = native_symbol;
  header.key().date = date;
  return std::make_unique<DatasetRecorderManager>(
      header,
      filename_suffix,
      config.filename(),
      config.recorder());
}

class FeedDumper : public coin2::strategy::IStrategy {
 public:
  explicit FeedDumper(
      const int64_t& date,
      const int64_t& end_time,
      const std::experimental::filesystem::path& recorder_config_file)
      : date_(date), end_time_(end_time), recorder_config_file_(recorder_config_file), book_print_level_(10) {
    LOG(INFO) << recorder_config_file;
    std::string config_str;
    str_read(recorder_config_file, &config_str);
    config_.ParseFromString(config_str);
    book_print_level_ = config_.recorder().book_print_level();
  }

  ~FeedDumper() { LOG(INFO) << "Bye"; }

  void onTradeFeed(const FeedUpdate& upd) override {
    ::presto2::FeedMessage feed_msg;
    ConvertFeedMessage(upd, book_print_level_, &feed_msg);
    Dump(upd, feed_msg);
  }

  void onBookFeed(const FeedUpdate& upd) override {
    ::presto2::FeedMessage feed_msg;
    ConvertFeedMessage(upd, book_print_level_, &feed_msg);
    Dump(upd, feed_msg);
  }

  void onOpenInterestFeed(const FeedUpdate& upd) override {
    ::presto2::FeedMessage feed_msg;
    ConvertFeedMessage(upd, book_print_level_, &feed_msg);
    Dump(upd, feed_msg);
  }

  void onIndexFeed(const FeedUpdate& upd) override {
    ::presto2::FeedMessage feed_msg;
    ConvertFeedMessage(upd, book_print_level_, &feed_msg);
    Dump(upd, feed_msg);
  }

  void Dump(const FeedUpdate& upd, const ::presto2::FeedMessage& feed) {
    papyrus::presto_feed_dataset::FeedRow feedrow;
    // std::string out;
    // feed.SerializeToString(&out);
    // feedrow.feed_message_.ParseFromString(out);
    // std::string out2;
    // feedrow.feed_message_.SerializeToString(&out2);
    feedrow.feed_message_.CopyFrom(feed);

    auto exchange = ExchangeTypeToString(upd.exchange());
    auto norm_symbol = upd.product().absolute_norm();
    Key key(exchange, norm_symbol);
    auto iter = manager_map_.find(key);
    if (iter != manager_map_.end()) {
      iter->second->ObserveFeedRow(feedrow);
    } else {
      config_.mutable_recorder()->set_archive_start_time(date_);
      config_.mutable_recorder()->set_archive_end_time(end_time_);
      manager_map_[key] = CreateDatasetRecorderManager(
          exchange,
          norm_symbol,
          upd.product().feed_native_symbol(),
          YmdFromTimestamp(date_),
          config_,
          norm_symbol);
      manager_map_[key]->ObserveFeedRow(feedrow);
    }
  }

  void Write() {
    for (const auto& iter : manager_map_) {
      iter.second->WriteDataset();
    }
  }

 private:
  using Key = std::pair<std::string, std::string>;
  int64_t date_;
  int64_t end_time_;
  std::experimental::filesystem::path recorder_config_file_;
  papyrus::presto_feed_dataset::DatasetRecorderManagerConfigProto config_;
  int32_t book_print_level_;
  std::unordered_map<Key, std::unique_ptr<DatasetRecorderManager>, boost::hash<Key>> manager_map_;
  ::coin2::exchange::base::symbology::ProductEncyclopedia product_cache_;
};

int main(int argc, char* argv[]) {
  DriverConfig app;

  cxxopts::Options opt("feed dumper", "feed dumper papyrus");
  opt.add_options()("recorder_config_filename", "", cxxopts::value<std::string>());
  AddDriverOptions(&opt);
  AddIntervalOptions(&opt);
  AddProductOptions(&opt);
  AddDateOption(&opt);
  AddFeedArchiveOptions(&opt);

  auto res = opt.parse(argc, argv);
  ParseDriverOptions(res, &app);
  ParseProductOptions(res, app.mutable_feed());
  // auto date = ParseDateOption(res);
  int64_t date = TimestampFromIso8601(res["date"].as<std::string>());
  int64_t end_time;
  {
    auto& interval = *app.mutable_feed()->mutable_common()->mutable_archive()->mutable_interval();
    interval.set_start(std::to_string(date));

    if (res.count("duration") != 0U) {
      auto duration_str = res["duration"].as<std::string>();
      auto duration = DurationFromString(duration_str);
      interval.set_end(std::to_string(date + duration));
    } else {
      interval.set_end(std::to_string(date + 24 * 60 * 60 * 1'000'000'000LL));
    }
    end_time =  std::stoull(interval.end());
  }

  ::coin2::base::InitLogging(argv[0], app.log());

  FeedDumper feed_dumper(
      date,
      end_time,
      res["recorder_config_filename"].as<std::string>()
      );
  Driver driver(app, &feed_dumper);
  driver.Run();

  feed_dumper.Write();
  return 0;
}
